# %%
# 初始化
import pandas as pd

import reader
from common import *

(year, month), (year_new, month_new) = year_month()
year_month_list = last_n_month(year, month)

FP_GONGDAN_LIST = [
    f"data/0-工单/家客工单导出_潮州_{year}-{month:02}.zip"
    for year, month in year_month_list
]
FP_GONGDAN_DANGYUE = (
    f"data/0-工单/家客工单导出_潮州_{year}-{month:02}.zip"
)
FP_HUGAN = f"data/0-互感/{year}{month:02}_互感数据.zip"
FP_GUANXIAN = f"data/0-管线/{year}{month:02}_管线数据.zip"
FP_KUANDAI_BI = (
    f"data/0-BI/{year_new}{month_new:02}_BI宽带月报.zip"
)
FP_ONU_GGL = (
    f"data/2-宽带装移机/{year}{month:02}_新装ONU管控报表清单.csv"
)
FP_ZHIJIAN = f"data/2-宽带装移机/{year}{month:02}_智能质检清单.xlsx"
FP_BAIMINGDAN = "data/2-宽带装移机/全市新增酒店类企宽明细白名单.xlsx"
FP_JIHE_MINGXI = (
    f"data/2-宽带装移机/{year}{month:02}_宽带开通稽核明细.zip"
)

check_files(
    FP_GONGDAN_LIST,
    FP_HUGAN,
    FP_ONU_GGL,
    FP_ZHIJIAN,
    FP_GUANXIAN,
    FP_BAIMINGDAN,
    FP_JIHE_MINGXI,
)

# %%
# 读取工单
usecols = (
    "工单号",
    "宽带帐号",
    "CRM工单号",
    "CRM业务流水号",
    "操作类型",
    "接入方式",
    "是否施工标识",
    "产品名称",
    "派单日期",
    "结单时间",
    "工单状态",
    "资源类型",
    "标准地址",
    "产品业务属性",
    "首次第三方复核结果",
    "光功率达标情况",
)
df = pd.concat(
    [
        reader.read_gongdan(f, usecols)
        for f in FP_GONGDAN_LIST
    ],
    ignore_index=True,
)
df_dy = reader.read_gongdan(FP_GONGDAN_DANGYUE, usecols)

# %%
# 筛选数据
# 操作类型 = 业务拆除
# 是否施工 = 普通开通施工
# 资源类型 = 自建宽带
# 工单状态 = 归档
type_mask = lambda df: (
    (df["操作类型"] == "业务拆除")
    | (
        (df["操作类型"] == "业务开通")
        & ((df["产品业务属性"] == "新增业务") | df["产品业务属性"].isna())
    )
    | ((df["操作类型"] == "业务移机") & (df["产品业务属性"] == "存量业务"))
)

other_mask = lambda df: (
    (
        (df["是否施工标识"].isna())
        | (df["是否施工标识"] == "普通开通施工")
        | (df["是否施工标识"] == "")
    )
    & (df["产品名称"].isin(["手机宽带", "宽带", "手机宽带基础产品"]))
    & (df["工单状态"] == "归档")
    # & (df["资源类型"] == "自建宽带")
)
df = df[type_mask(df)]
df = df[other_mask(df)]
df_dy = df_dy[type_mask(df_dy)]
df_dy = df_dy[other_mask(df_dy)]

df_dy_cj = df_dy[df_dy["操作类型"] == "业务拆除"]
df_dy_kt = df_dy[df_dy["操作类型"] == "业务开通"]
df_dy_yj = df_dy[df_dy["操作类型"] == "业务移机"]
df_dy = df_dy[df_dy["操作类型"].isin(["业务开通", "业务移机"])]


# %%
# 当月开通次月0流量核查

df_bi = reader.read_csv(
    FP_KUANDAI_BI,
    columns=("宽带号码", "T月流量"),
    dtype={"宽带号码": "string", "T月流量": "float"},
)

df_bi.dropna(inplace=True)

mask = df_dy["宽带帐号"].isin(df_bi[df_bi["T月流量"] > 5]["宽带号码"])
df_dy.loc[mask, "当月开通次月0流量核查"] = S_PASSED
df_dy.loc[~mask, "当月开通次月0流量核查"] = S_NOT_PASSED
df_dy.loc[
    df_dy.宽带帐号.str.endswith("a"), "当月开通次月0流量核查"
] = S_BYPASSED

# %%
# 90天同址同帐号先拆后装核查
df_tmp = df_dy_kt.merge(
    df.loc[
        df["操作类型"] == "业务拆除",
        [
            "宽带帐号",
            "标准地址",
            "派单时间",
        ],
    ],
    on=["宽带帐号", "标准地址"],
)

mask = df_dy["工单号"].isin(
    df_tmp[
        df_tmp["派单时间_x"] - df_tmp["派单时间_y"]
        < pd.Timedelta(days=90)
    ]["工单号"]
)
df_dy["90天同址同帐号先拆后装核查"] = S_PASSED
df_dy.loc[
    df_dy["操作类型"] != "业务开通", "90天同址同帐号先拆后装核查"
] = S_BYPASSED
df_dy.loc[mask, "90天同址同帐号先拆后装核查"] = S_NOT_PASSED
# 超长
mask = (df_dy["操作类型"] == "业务开通") & (
    (df_dy["归档时间"] - df_dy["派单时间"]) > pd.Timedelta(days=90)
)
df_dy.loc[mask, "90天同址同帐号先拆后装核查"] = S_NOT_PASSED


# %%
# 当月同帐号先装后拆核查
df_tmp = df_dy_kt.merge(
    df_dy_cj[["宽带帐号", "派单时间"]],
    on=["宽带帐号"],
)

mask = df_dy["工单号"].isin(
    df_tmp[df_tmp["派单时间_x"] < df_tmp["派单时间_y"]]["工单号"]
)

df_dy["当月同帐号先装后拆核查"] = S_PASSED
df_dy.loc[
    df_dy["操作类型"] != "业务开通", "当月同帐号先装后拆核查"
] = S_BYPASSED
df_dy.loc[mask, "当月同帐号先装后拆核查"] = S_NOT_PASSED

# %%
# 90天内同帐号先装后移核查
df_tmp = df_dy_yj.merge(
    df.loc[
        df["操作类型"] == "业务开通",
        [
            "宽带帐号",
            "派单时间",
        ],
    ],
    on=["宽带帐号"],
)

mask = df_dy["工单号"].isin(
    df_tmp[
        df_tmp["派单时间_x"] - df_tmp["派单时间_y"]
        < pd.Timedelta(days=90)
    ]["工单号"]
)

df_dy["90天内同帐号先装后移核查"] = S_PASSED
df_dy.loc[
    df_dy["操作类型"] != "业务移机", "90天内同帐号先装后移核查"
] = S_BYPASSED
df_dy.loc[mask, "90天内同帐号先装后移核查"] = S_NOT_PASSED

# %%
# 90内同帐号多次移机核查
df_tmp = df_dy_yj.merge(
    df.loc[
        df["操作类型"] == "业务移机",
        [
            "工单号",
            "宽带帐号",
            "派单时间",
        ],
    ],
    on=["宽带帐号"],
)

mask = df_dy["工单号"].isin(
    df_tmp[
        (df_tmp["工单号_x"] != df_tmp["工单号_y"])
        & (
            df_tmp["派单时间_x"] - df_tmp["派单时间_y"]
            < pd.Timedelta(days=90)
        )
    ]["工单号_x"]
)

df_dy["90内同帐号多次移机核查"] = S_PASSED
df_dy.loc[
    df_dy["操作类型"] != "业务移机", "90内同帐号多次移机核查"
] = S_BYPASSED
df_dy.loc[mask, "90内同帐号多次移机核查"] = S_NOT_PASSED

# %%
# 当月相同联系电话装机工单核查（不稽核）
df_dy["当月相同联系电话装机工单核查"] = S_BYPASSED

# %%
# 当月新增一址多户工单核查
df_guanxian = reader.read_csv(
    FP_GUANXIAN,
    columns=["宽带号码", "七级地址ID"],
)
guanxian_err_addr = df_guanxian.groupby("七级地址ID").filter(
    lambda x: len(x) >= 5
)["宽带号码"]
df_dy["当月新增一址多户工单核查"] = S_PASSED
df_dy.loc[
    df_dy["宽带帐号"].isin(guanxian_err_addr), "当月新增一址多户工单核查"
] = S_NOT_PASSED

# %%
# 登录端口异常工单核查
df_dy["登录端口异常工单核查"] = S_PASSED

df_hugan = reader.read_csv(
    FP_HUGAN,
    columns=(
        "宽带号码",
        "ONU下联端口",
        "所属ONU设备名称",
        "错误类型",
    ),
)
df_hugan_err_port = df_hugan.groupby(
    ["ONU下联端口", "所属ONU设备名称"]
).filter(lambda x: len(x) >= 2)["宽带号码"]
df_dy.loc[
    df_dy["宽带帐号"].isin(df_hugan_err_port), "登录端口异常工单核查"
] = S_NOT_PASSED


# %%
# 当月登录端口与资源系统不一致核查
df_dy["当月登录端口与资源系统不一致核查"] = S_NOT_PASSED
df_hugan_ok_ziyuan = df_hugan[df_hugan["错误类型"] == "正确资源"][
    "宽带号码"
]
df_dy.loc[
    df_dy["宽带帐号"].isin(df_hugan_ok_ziyuan),
    "当月登录端口与资源系统不一致核查",
] = S_PASSED

# %%
# 工单规范性质检
df_znzj = pd.read_excel(
    FP_ZHIJIAN,
    usecols=("工单编号", "业务类型"),
    dtype="string",
    skiprows=1,
)
df_znzj = df_znzj[df_znzj["业务类型"] == "宽带"]["工单编号"]
mask = df_dy["工单号"].isin(df_znzj) | (
    df_dy["首次第三方复核结果"] == S_NOT_PASSED
)
df_dy["工单规范性质检"] = S_PASSED
df_dy.loc[mask, "工单规范性质检"] = S_NOT_PASSED


# %%
# 弱光核查
dtype_dict = {
    "工单号": "string",
    "操作类型": "string",
    "产品类型": "string",
    "光功率是否达标": "string",
    "整改后光功率值": "float",
}
df_ggl = pd.read_csv(
    FP_ONU_GGL,
    usecols=dtype_dict.keys(),
    encoding="gbk",
    dtype=dtype_dict,
)

ggl_ok = df_ggl.loc[
    (df_ggl["产品类型"] == "家客开通")
    & (
        (df_ggl["光功率是否达标"] == "是")
        | (df_ggl["整改后光功率值"] >= -25)
    ),
    "工单号",
]
df_dy["弱光核查"] = S_NOT_PASSED
df_dy.loc[
    df_dy["工单号"].isin(ggl_ok),
    "弱光核查",
] = S_PASSED
df_dy.loc[
    (df_dy["接入方式"] == "FTTB"),
    "弱光核查",
] = S_BYPASSED

# 新装ONU管控报表清单中无对应清单
mask = (
    (df_dy.接入方式 == "FTTH")
    & (~df_dy.工单号.isin(df_ggl.工单号))
    & (df_dy.光功率达标情况 == "已达标")
)
df_dy.loc[mask, "弱光核查"] = S_PASSED

# %%
# 酒店类白名单核查
df_bmd = pd.read_excel(
    FP_BAIMINGDAN,
    usecols=("新增月份", "宽带账号"),
    dtype="string",
)

df_bmd = df_bmd[(df_bmd["新增月份"] == f"{year}{month:02}")]

df_dy["酒店类白名单核查"] = S_NO
df_dy.loc[
    df_dy["宽带帐号"].isin(df_bmd["宽带账号"]), "酒店类白名单核查"
] = S_YES

# %%
# 超长工单核查
df_dy["超长工单核查"] = S_PASSED
mask = df_dy["归档时间"] - df_dy["派单时间"] > pd.Timedelta(days=90)
df_dy.loc[mask, "超长工单核查"] = S_NOT_PASSED

# %%
# 省公司稽核报表
columns = (
    "地市编码",
    "地市名称",
    "账期",
    "工单号",
    "CRM单号",
    "宽带号码",
    "操作类型",
    "派单时间",
    "归档日期",
    "接入模式",
    "7级地址ID",
    "网格名称",
    "班组",
    "代维账号",
    "接单人",
    "客户类型",
    "系统稽核结果",
    "当月开通次月0流量核查",
    "90天同址同帐号先拆后装核查",
    "当月同帐号先装后拆核查",
    "90天内同帐号先装后移核查",
    "90内同帐号多次移机核查",
    "当月相同联系电话装机工单核查",
    "当月新增一址多户工单核查",
    "登录端口异常工单核查",
    "当月登录端口与资源系统不一致核查",
    "工单规范性质检",
    "弱光核查",
)

df_sgs = reader.read_csv(
    FP_JIHE_MINGXI,
    columns=columns,
)


# 报告汇总
# CRM业务流水号、资源类型、地市稽核结果、地市与系统稽核不一致原因
tmp = df_sgs.merge(
    df_dy, on="工单号", how="left", suffixes=("_sgs", "_cz")
)
df_sgs["CRM业务流水号"] = tmp["CRM业务流水号"]
df_sgs["资源类型"] = tmp["资源类型"]

# %%
# 地市稽核结果
CHECK_COLUMNS = [
    "当月开通次月0流量核查",
    "90天同址同帐号先拆后装核查",
    "当月同帐号先装后拆核查",
    "90天内同帐号先装后移核查",
    "90内同帐号多次移机核查",
    "当月相同联系电话装机工单核查",
    "当月新增一址多户工单核查",
    "登录端口异常工单核查",
    "当月登录端口与资源系统不一致核查",
    "工单规范性质检",
    "弱光核查",
]

columns = [f"{col}_cz" for col in CHECK_COLUMNS]

not_passed = tmp[columns].isin([S_NOT_PASSED]).any(axis=1)
df_sgs["地市稽核结果"] = S_PASSED
df_sgs.loc[not_passed, "地市稽核结果"] = S_NOT_PASSED
# %%
# 酒店类白名单核查
df_white_list = pd.read_excel(
    FP_BAIMINGDAN,
    usecols=["宽带账号"],
    dtype="string",
)

df_sgs["酒店类白名单核查"] = S_NO
df_sgs.loc[
    df_sgs["宽带号码"].isin(df_white_list["宽带账号"]), "酒店类白名单核查"
] = S_YES

mask = (df_sgs["酒店类白名单核查"] == S_YES) & (
    df_sgs["当月开通次月0流量核查"] == S_NOT_PASSED
)
df_sgs.loc[mask, "地市稽核结果"] = S_PASSED

# %%
# CRM业务流水号
df_sgs.loc[
    df_sgs["CRM业务流水号"].isna(), "地市稽核结果"
] = S_NOT_PASSED

# %%
# 地市与系统稽核不一致原因
def explain(x):
    reasons = []
    for col in CHECK_COLUMNS:
        if (x[f"{col}_cz"] == S_NOT_PASSED) and (
            x[f"{col}_sgs"] == S_PASSED
        ):
            reasons.append(col)

    return ";".join(reasons)


df_tmp = df_sgs.merge(
    df_dy, on="工单号", how="left", suffixes=("_sgs", "_cz")
)
df_sgs["地市与系统稽核不一致原因"] = df_tmp.apply(explain, axis=1)
df_sgs.loc[~df_sgs.工单号.isin(df_dy.工单号), "市缺"] = S_YES

df_sgs.to_excel(
    f"report/2-宽带装移机/{year}{month:02}_宽带开通工单地市铁通核对表.xlsx",
    index=False,
)

# %%
# 资源类型核查
df_dy["资源类型核查"] = S_NOT_PASSED
df_dy.loc[df_dy.资源类型 == "自建宽带", "资源类型核查"] = S_PASSED


# %%
# 地市稽核结果
CHECK_COLUMNS += ["资源类型核查"]
not_passed = (df_dy[CHECK_COLUMNS] == S_NOT_PASSED).any(
    axis=1
)
df_dy["地市稽核结果"] = S_PASSED
df_dy.loc[not_passed, "地市稽核结果"] = S_NOT_PASSED

mask = (df_dy["酒店类白名单核查"] == S_YES) & (
    df_dy["当月开通次月0流量核查"] == S_NOT_PASSED
)
df_dy.loc[mask, "地市稽核结果"] = S_PASSED


# %%
# 宽带装移机工单地市稽核明细表
df_dy = df_dy.merge(
    df_sgs[["工单号", "系统稽核结果"]], on="工单号", how="left"
)
df_dy.loc[~df_dy.工单号.isin(df_sgs.工单号), "省缺"] = S_YES
df_dy.to_excel(
    f"report/2-宽带装移机/{year}{month:02}_宽带装移机工单地市稽核明细表.xlsx",
    index=False,
)
df_dy.to_pickle(
    f"pickle/{year}{month:02}_宽带装移机工单地市稽核明细表.pkl"
)

# %%
# 完成
